package com.hugang.juc.blockingqueue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者消费者示例 —— 使用 volatile/cas/atomicInteger/blockingQueue/线程交互/原子引用
 *
 * @author hg
 * @date 2020/8/25 16:19
 */
public class ProdConsumeBlockingQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        ArrayBlockingQueue<Integer> integers = new ArrayBlockingQueue<>(10);
        MyResource myResource = new MyResource(integers);

        new Thread(()->{
                try {
                    myResource.prod();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
        }, "prod").start();

        new Thread(()->{
                try {
                    myResource.consume();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
        }, "consume").start();

        TimeUnit.SECONDS.sleep(50);

        myResource.stop();
    }

}

/**
 * 资源类
 */
class MyResource{

    //标志位，表示是否进行生产消费活动
    private volatile boolean flag = true;
    //默认为0
    private AtomicInteger integer = new AtomicInteger();
    //阻塞队列
    BlockingQueue<Integer> blockingQueue = null;

    public MyResource(BlockingQueue<Integer> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    /**
     * 生产方法
     */
    public void prod() throws InterruptedException {
        //flag为true表示开始生产消费活动
        Integer num = null;
        boolean temp = false;
        while (flag){
            num = integer.incrementAndGet();
            temp = blockingQueue.offer(num,2000, TimeUnit.MILLISECONDS);
            if (temp){
                System.out.println(Thread.currentThread().getName() +" ---> 生产了：" + num);
            } else {
                System.out.println(Thread.currentThread().getName() +" ---> 队列满了导致生产活动超时，生产失败");
            }
            TimeUnit.MILLISECONDS.sleep(100);
        }
        System.out.println(Thread.currentThread().getName() +" ---> 流程被叫停，生产活动终止。");

    }

    /**
     * 消费方法
     */
    public void consume() throws InterruptedException {
        //flag为true表示开始生产消费活动
        Integer num = null;
        while (flag){
            num = blockingQueue.poll(2000, TimeUnit.MILLISECONDS);
            if (num == null){
                //等于空代表从阻塞队列获取数据，在等待时间过去之后依旧未获取到数据
                System.out.println(Thread.currentThread().getName() +" ---> 队列为空导致消费活动超时，消费失败");
            } else {
                System.out.println(Thread.currentThread().getName() +" ---> 消费了：" + num);
            }
            TimeUnit.SECONDS.sleep(2);
        }
        System.out.println(Thread.currentThread().getName() +" ---> 流程被叫停，消费活动终止。");
    }

    /**
     * 停止活动
     */
    public void stop(){
        this.flag = false;
    }

}
